package com.shujia.flink.window

import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
import org.apache.flink.util.Collector

object Demo3Window {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val students: DataStream[String] = env.readTextFile("data/students.txt")

    /**
      * 统计最近每个班级10个学生的平均年龄，每隔一个学生统计一次
      */

    val kvDS: DataStream[(String, Int)] = students.map(student =>{
      val split: Array[String] = student.split(",")
      (split(4),split(2).toInt)
    })

    /**
      * process: 在窗口之后对窗口内的数据进行处理
      *
      */
    val avgAgg: DataStream[(String, Double)] = kvDS.keyBy(_._1)
      .countWindow(10,1)
      .process(new MyProcessWindowFunction)

    avgAgg.print()

    env.execute()
  }
}
class MyProcessWindowFunction extends ProcessWindowFunction[(String, Int), (String, Double), String, GlobalWindow] {
  /**
    *
    *
    * process： 每一个窗口中的每一个key运行一次
    *
    * @param key      班级
    * @param context  上下文对象
    * @param elements 班级在这个窗口中的所有数据
    * @param out      将数据发送到下游
    */
  override def process(key: String, context: Context, elements: Iterable[(String, Int)], out: Collector[(String, Double)]): Unit = {


    //取出年龄
    val ages: List[Int] = elements.toList.map(_._2)
    //计算平均年龄
    val avgAge: Double = ages.sum.toDouble / ages.length

    //将数据发送到下游
    out.collect((key, avgAge))

  }

}